package com.yanggu.flink.datastream_api.multi_stream_transform.combine_stream

import com.yanggu.flink.datastream_api.pojo.Event
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.time.Duration

/**
 * union之后的Watermark如何计算水位线呢？和之前的分区Watermark一致, 仍然是取上游传递的最小的Watermark,
 * 只有当最小的更新了之后，当前算子的Watermark才会更新
 */
object UnionDemo02 {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val dataStream1 = env
      .socketTextStream("localhost", 7777)
      .map(line => {
        val dataArray = line.split(",")
        Event(dataArray(0), dataArray(1), dataArray(2).toLong)
      })
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forBoundedOutOfOrderness(Duration.ofSeconds(2L))
        .withTimestampAssigner(new SerializableTimestampAssigner[Event] {
          override def extractTimestamp(element: Event, recordTimestamp: Long) = element.timestamp
        })
      )

    val dataStream2 = env
      .socketTextStream("localhost", 8888)
      .map(line => {
        val dataArray = line.split(",")
        Event(dataArray(0), dataArray(1), dataArray(2).toLong)
      })
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forBoundedOutOfOrderness(Duration.ofSeconds(2L))
        .withTimestampAssigner(new SerializableTimestampAssigner[Event] {
          override def extractTimestamp(element: Event, recordTimestamp: Long) = element.timestamp
        })
      )

    dataStream1.print("DataStream1>>>")
    dataStream2.print("DataStream2>>>")

    dataStream1
      .union(dataStream2)
      .process((_: Event, ctx: ProcessFunction[Event, String]#Context, out: Collector[String]) => {
        out.collect(s"水位线: ${ctx.timerService().currentWatermark()}")
      })
      .print("union后的Watermark>>>")

    env.execute()

  }

}
